package net.opentsdb.tsd;

import com.stumbleupon.async.Callback;
import com.stumbleupon.async.Deferred;
import net.opentsdb.core.IncomingDataPoint;
import net.opentsdb.core.TSDB;
import net.opentsdb.core.Tags;
import net.opentsdb.utils.Config;
import org.hbase.async.HBaseException;
import org.hbase.async.PleaseThrottleException;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.jboss.netty.handler.codec.http.HttpResponseStatus;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author Hujf
 * @title: PutDataPointRpc
 * @date 2021-09-17 16:55
 * @description: 插入数据的RPC处理类
 */
public class PutDataPointRpc implements TelnetRpc, HttpRpc {

    /**
     * Whether or not to send error messages back over telnet
     */
    private final boolean send_telnet_errors;
    protected static final AtomicLong telnet_requests = new AtomicLong();

    protected static final AtomicLong http_requests = new AtomicLong();

    protected static final AtomicLong inflight_exceeded = new AtomicLong();
    protected static final AtomicLong hbase_errors = new AtomicLong();
    protected static final AtomicLong illegal_arguments = new AtomicLong();
    protected static final AtomicLong unknown_errors = new AtomicLong();


    public PutDataPointRpc(Config config) {
        send_telnet_errors = config.getBoolean("tsd.rpc.telnet.return_errors");
    }


    /**
     * 执行http方法
     */
    @Override
    public void execute(TSDB tsdb, HttpQuery query) {
        http_requests.incrementAndGet();

        if (query.method() != HttpMethod.POST) {
            throw new BadRequestException(HttpResponseStatus.METHOD_NOT_ALLOWED,
                    "Method not allowed", "The HTTP method [" + query.method().getName() +
                    "] is not permitted for this endpoint");
        }

        List<IncomingDataPoint> dps;

        dps = query.serializer().parsePutV1(IncomingDataPoint.class, HttpJsonSerializer.TR_INCOMING);

        try {
            processDataPoint(tsdb, query, dps);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 执行数据处理
     *
     * @param tsdb
     * @param query
     * @param dps
     */
    private void processDataPoint(TSDB tsdb, HttpQuery query, List<IncomingDataPoint> dps) throws Exception {
        int queued = 0;
        if (dps.size() < 1) {
            throw new BadRequestException("没有数据");
        }
        final HashMap<String, String> query_tags = new HashMap<>();

        final AtomicBoolean sending_response = new AtomicBoolean();
        sending_response.set(false);


        final boolean synchronous = query.hasQueryStringParam("sync");
        final List<Deferred<Boolean>> deferreds = synchronous ? new ArrayList<Deferred<Boolean>>(dps.size()) : null;

        List detail = new ArrayList<>();

        for (final IncomingDataPoint dp : dps) {
            DataPointType type = null;
            type = DataPointType.PUT;


            //两个回调  成功和失败
            final class SuccessCB implements Callback {

                @Override
                public Object call(Object o) throws Exception {
                    return null;
                }
            }

            final class PutErrorback implements Callback<Object, Exception> {

                @Override
                public Object call(Exception e) throws Exception {
                    e.printStackTrace();
                    return null;
                }
            }

            if (dp == null) {
                continue;
            }


            if (!dp.validate(detail)) {
                //  continue;
            }

            //执行单点插入
            Deferred<Boolean> deferred = null;

            if (type == DataPointType.HISTOGRAM) {
                deferred = null;
            } else {
                if (Tags.looksLikeInteger(dp.getValue())) {
                    switch (type) {
                        case ROLLUP:
                            break;
                        default:
                            deferred = tsdb.addPoint(dp.getMetric(), dp.getTimestamp(), Long.parseLong(dp.getValue()), dp.getTags())
                                    .addCallback(new SuccessCB())
                                    .addCallback(new PutErrorback());
                    }
                } else {
                    switch (type) {
                        case ROLLUP:
                            break;
                        default:
                            deferred = tsdb.addPoint(dp.getMetric(), dp.getTimestamp(),
                                    (Tags.fitsInFloat(dp.getValue()) ?
                                            Float.parseFloat(dp.getValue()) :
                                            Double.parseDouble(dp.getValue())), dp.getTags())
                                    .addCallback(new SuccessCB())
                                    .addErrback(new PutErrorback());
                    }
                }
            }
            ++queued;
            if (synchronous) {
                deferreds.add(deferred);
            }
        }

        class GroupCB implements Callback<Object, ArrayList> {
            int queued;
            public GroupCB(int queued) {
                this.queued = queued;
            }

            @Override
            public Object call(ArrayList arg) throws Exception {
                query.sendReply("".getBytes(StandardCharsets.UTF_8));
                return null;
            }
        }
        new GroupCB(queued).call(new ArrayList<>());
    }


    /**
     * The type of data point we're writing.
     *
     * @since 2.4
     */
    public enum DataPointType {
        PUT("put"),
        ROLLUP("rollup"),
        HISTOGRAM("histogram");

        private final String name;

        DataPointType(final String name) {
            this.name = name;
        }

        @Override
        public String toString() {
            return name;
        }
    }

    @Override
    public Deferred<Object> execute(TSDB tsdb, Channel chan, String[] cmd) {
        telnet_requests.incrementAndGet();
        final DataPointType type;
        String command = cmd[0].toLowerCase();
        if (command.equals("put")) {
            type = DataPointType.PUT;
        } else {
            throw new IllegalArgumentException("Unrecognized command: " + command);
        }

        class PutErrback implements Callback<Object, Exception> {

            @Override
            public Object call(Exception arg) {
                String errmsg = null;
                if (arg instanceof PleaseThrottleException) {
                    if (send_telnet_errors) {
                        errmsg = type + ": Please throttle writes: " + arg.getMessage() + '\n';
                    }
                    inflight_exceeded.incrementAndGet();
                } else {
                    if (send_telnet_errors) {
                        errmsg = type + ": HBase error: " + arg.getMessage() + '\n';
                    }
                    if (arg instanceof HBaseException) {
                        hbase_errors.incrementAndGet();
                    } else if (arg instanceof IllegalArgumentException) {
                        illegal_arguments.incrementAndGet();
                    } else {
                        unknown_errors.incrementAndGet();
                    }
                }

                handleStorageException(tsdb, getDataPointFromString(tsdb, cmd), arg);
                return true;
            }
        }

        final class SuccessCB implements Callback<Object, Object> {

            @Override
            public Object call(Object arg) throws Exception {
                return null;
            }
        }

        return importDataPoint(tsdb, cmd);
    }

    //解析每个命令
    private Deferred<Object> importDataPoint(TSDB tsdb, String[] cmd) {
        cmd[0] = null;
        if (cmd.length < 5) {
            throw new IllegalArgumentException("not enough arguments" + " ,need least 4 ,got" + (cmd.length));
        }
        final String metric = cmd[1];

        if (metric.length() <= 0) {
            throw new IllegalArgumentException("empty metric name");
        }

        final long timestamp;
        if (cmd[2].contains(".")) {
            timestamp = Tags.parseLong(cmd[2].replace(".", ""));
        } else {
            timestamp = Tags.parseLong(cmd[2]);
        }

        if (timestamp <= 0) {
            throw new IllegalArgumentException("invalid timestamp: " + timestamp);
        }

        final String value = cmd[3];
        if (value.length() <= 0) {
            throw new IllegalArgumentException("empty value");
        }

        final Map<String, String> tags = new HashMap<String, String>();
        for (int i = 4; i < cmd.length; i++) {
            if (!cmd[i].isEmpty()) {
                Tags.parse(tags, cmd[i]);
            }
        }

        if (Tags.looksLikeInteger(value)) {
            return tsdb.addPoint(metric, timestamp, Tags.parseLong(value), tags);
        } else if (Tags.fitsInFloat(value)) {  // floating point value
            return tsdb.addPoint(metric, timestamp, Float.parseFloat(value), tags);
        } else {
            return tsdb.addPoint(metric, timestamp, Double.parseDouble(value), tags);
        }

    }

    //解析命令 得到数据插入实体类
    private IncomingDataPoint getDataPointFromString(TSDB tsdb, String[] command) {
        IncomingDataPoint dp = new IncomingDataPoint();
        dp.setMetric(command[1]);
        if (command[2].contains(".")) {
            dp.setTimestamp(Tags.parseLong(command[2].replace(".", "")));
        } else {
            dp.setTimestamp(Tags.parseLong(command[2]));
        }

        dp.setValue(command[3]);

        HashMap<String, String> tags = new HashMap();
        for (int i = 4; i < command.length; i++) {
            if (!command[i].isEmpty()) {
                Tags.parse(tags, command[i]);
            }
        }
        dp.setTags(tags);
        return dp;
    }

    /**
     * Passes a data point off to the storage handler plugin if it has been
     * configured.
     *
     * @param tsdb The TSDB from which to grab the SEH plugin
     * @param dp   The data point to process
     * @param e    The exception that caused this
     */
    void handleStorageException(final TSDB tsdb, final IncomingDataPoint dp,
                                final Exception e) {
        final StorageExceptionHandler handler = tsdb.getStorageExceptionHandler();
        if (handler != null) {
            handler.handleError(dp, e);
        }
    }
}
